/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.io.OutputStream;

import jvs.peer.util.Log;


/**
 * @author qiangli
 *
 */
public class MessageOutputStream extends OutputStream {
	private Channel channel = null;

	private byte[] buf = null;
	
	private int pos = 0;
	private int seqno = 0; //messages sent
	private int counter = 0; //bytes sent
	
	public MessageOutputStream(Channel channel) {
		this.channel = channel;
		this.buf = new byte[channel.getBufferSize()];
	}

	@Override
	public void write(int b) throws IOException {
		buf[pos++] = (byte)b;
		if (pos >= buf.length) {
			send(buf,buf.length,MessageConstant.data);
			pos = 0;
		}
	}

	@Override
	synchronized public void close() throws IOException {
		if (buf != null) {
			flush();
			send(MessageConstant.close);
		}
		//
		buf = null;
		channel = null;
	}

	@Override
	public void flush() throws IOException {
		if (pos > 0) {
			send(buf,pos,MessageConstant.data);
			pos = 0;
		}
	}
	
	public void open() throws IOException {
		send(MessageConstant.open);
	}
	
	protected void send(int ioscmd) throws IOException {
		send(null,0,ioscmd);
	}
	
	protected void send(byte[] buf, int len, int ioscmd) throws IOException {
		if (channel == null) {
			throw new IOException("Stream closed ioscmd:  "+ ioscmd);
		}
		
		Message msg = new Message();
		try {
			msg.setIoscmd(ioscmd);
			
			msg.setData(buf);
			msg.setSize(len);
			
			msg.setSeqno(seqno);

			channel.send(msg);	
			seqno++;
			
			counter += len;	
		} catch (Exception e) {
			if (Log.DEBUG) {
				e.printStackTrace();
			}
			throw new IOException(e.getMessage());
		}

		if(Log.DEBUG) {
			Log.log(this, Thread.currentThread().getId()+"@@@@ "+channel.getId()+" out seqno: "+msg.getSeqno()+" counter: "+counter);
		}
		
	}

}
